package com.chukun.gateway.core.parallel.disruptor;

import com.chukun.gateway.core.parallel.ParallelHandlerQueue;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.lmax.disruptor.*;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author chukun
 * @version 1.0.0
 * @description 基于Disruptor实现的多生产多消费者无锁队列处理类
 * @createTime 2023/12/23 20:03
 */
public class DisruptorParallelQueueHandler<E> implements ParallelHandlerQueue<E> {

    private RingBuffer<BufferHolder<E>> ringBuffer;

    private final WorkerPool<BufferHolder<E>> workerPool;

    private final ExecutorService executorService;

    private final EventTranslatorOneArg<BufferHolder<E>,E> eventTranslator;

    private final EventListener<E> eventListener;


    public DisruptorParallelQueueHandler(DisruptorHandlerBuilder<E> builder) {
        this.executorService = Executors.newFixedThreadPool(builder.getThreads(),
                new ThreadFactoryBuilder().setNameFormat("ParallelQueueHandler"+builder.getNamePrefix()+"-pool-%d").build());

        this.eventTranslator = new BufferHolderEventTranslator<E>();

        this.eventListener = builder.getEventListener();

        //创建RingBuffer
        RingBuffer<BufferHolder<E>> ringBuffer = RingBuffer.create(builder.getProducerType(),
                new BufferHolderEventFactory<E>(),
                builder.getBufferSize(),
                builder.getWaitStrategy());

        //通过RingBuffer 创建屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        //创建多个消费者组
        WorkHandler<BufferHolder<E>>[] workHandlers = new WorkHandler[builder.getThreads()];
        for (int i = 0; i< workHandlers.length;i++){
            workHandlers[i] = new BufferHolderWorkHandler<>(eventListener);
        }

        //创建多消费者线程池
        WorkerPool<BufferHolder<E>> workerPool = new WorkerPool<BufferHolder<E>>(ringBuffer,
                sequenceBarrier,
                new BufferHolderExceptionHandler(),
                workHandlers);
        //设置多消费者的Sequence序号，主要用于统计消费进度，
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        this.workerPool = workerPool;
    }

    @Override
    public void add(E event) {
        if(this.ringBuffer == null){
            process(this.eventListener,new IllegalStateException("ParallelQueueHandler is close"),event);
        }
        try {
            ringBuffer.publishEvent(this.eventTranslator, event);
        }catch (NullPointerException e){
            process(this.eventListener,new IllegalStateException("ParallelQueueHandler is close"),event);
        }
    }


    @Override
    @SafeVarargs
    public final void add(E... events) {
        if(ringBuffer == null){
            process(this.eventListener,new IllegalStateException("ParallelQueueHandler is close"),events);
        }
        try {
            ringBuffer.publishEvents(this.eventTranslator,events);
        }catch (NullPointerException e){
            process(this.eventListener,new IllegalStateException("ParallelQueueHandler is close"),events);
        }
    }

    @Override
    public boolean tryAdd(E event) {
        if(this.ringBuffer == null){
            return false;
        }
        try {
            return  ringBuffer.tryPublishEvent(this.eventTranslator,event);
        }catch (NullPointerException e){
            return false;
        }
    }


    @Override
    @SafeVarargs
    public final boolean tryAdd(E... events) {
        if(this.ringBuffer == null){
            return false;
        }
        try {
            return  ringBuffer.tryPublishEvents(this.eventTranslator,events);
        }catch (NullPointerException e){
            return false;
        }
    }

    @Override
    public void start() {
        this.ringBuffer = workerPool.start(executorService);
    }

    @Override
    public void shutDown() {
        if(this.ringBuffer ==null){
            return;
        }
        if(workerPool != null){
            workerPool.drainAndHalt();
        }
        if(executorService != null){
            executorService.shutdown();
        }
    }

    @Override
    public boolean isShutDown() {
        return this.ringBuffer ==null;
    }

    private static <E> void process(EventListener<E> listener,Throwable e,E event){
        listener.onException(e,-1,event);
    }

    @SafeVarargs
    private static <E> void process(EventListener<E> listener, Throwable e, E... events){
        for(E event: events){
            process(listener,e,event);
        }
    }
}
